package com.dsk.connector.sink.common.utils;

import cn.hutool.core.date.DateUtil;
import com.dsk.connector.commons.conf.TableMapping;
import com.dsk.connector.commons.trans.FieldSchema;
import com.dsk.connector.commons.trans.RecordDml;
import org.apache.commons.lang3.StringUtils;

import java.util.Date;
import java.util.Map;
import java.util.function.Function;

public class SinkUtils {
    /**
     * 根据数据计算要插入或更新的目标表
     * @param tableMapping
     * @param record
     * @return
     */
    public static String getTargetTable(TableMapping tableMapping, RecordDml record){
        if(tableMapping.isSplitTable()){
            Function<String,String> getTableSuffix=valueStr->{
                String suffix=valueStr;
                if(tableMapping.getSplitFieldLength()>0 && tableMapping.getSplitFieldLength()<suffix.length()){
                    suffix=suffix.substring(0,tableMapping.getSplitFieldLength());
                }
                return String.format("%s_%s",tableMapping.getTargetTable(),suffix);
            };
            Map<String,Object> valueMap=record.getPayloadValues();
            for(String field:tableMapping.getSplitFields()){
                if(valueMap.get(field)!=null && StringUtils.isNotBlank(valueMap.get(field).toString())){
                    Object fieldValue=valueMap.get(field);
                    FieldSchema schema=record.getPayloadSchemaMap().get(field);
                    if("io.debezium.time.Timestamp".equals(schema.getName())){
                        Date dateValue=DateUtil.parse(fieldValue.toString(),"yyyy-MM-dd HH:mm:ss");
                        String dateStr=DateUtil.format(dateValue,"yyyyMMddHHmmss");
                        return getTableSuffix.apply(dateStr);
                    }else if("io.debezium.time.Date".equals(schema.getName())){
                        Date dateValue=DateUtil.parse(fieldValue.toString(),"yyyy-MM-dd");
                        String dateStr=DateUtil.format(dateValue,"yyyyMMdd");
                        return getTableSuffix.apply(dateStr);
                    }else{
                        return String.format("%s_%s",tableMapping.getTargetTable(),fieldValue.toString());
                    }

                }
            }
            //针对没有符合条件数据的情况，则插入到原表中去
            return tableMapping.getTargetTable();
        }else{
            return tableMapping.getTargetTable();
        }

    }

}
